package com.wuwangfu.window.event;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @Author jcshen
 * @Date 2023-02-24
 * @PackageName:com.wuwangfu.window.event
 * @ClassName: EventTimeTumbleWindowAll
 * @Description:
 * @Version 1.0.0
 *
 * 不分组，按照event Time划分滚动窗口
 *
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/windows/#tumbling-windows
 */
public class EventTimeTumbleWindowAll {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        /*旧API 需要设置窗口划分的时间标准*/
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        /**
         * 1677168000000,1
         * 1677168003000,2
         * 1677168004400,2
         * 1677168004998,2
         * 1677168005000,2
         *
         * 1677168009998,2
         * 1677168009999,2
         *
         * 第一个窗口是 [1677168000000,1677168005000) 严格的说 [1677168000000,1677168004999]
         * 第二个窗口是 [1677168005000,1677168010000)
         *
         * 当前分区中，数据携带的最大EventTime - 乱序延迟时间 >= 窗口的结束时间；就会触发该窗口
         * 1677168005000 - 0 >= 1677168004999
         * 1677168009999 -0 >= 1677168004999
         */
        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);
        /*旧API 提取数据中的时间，将时间转成精确到毫秒的long类型，生成watermark*/
        //调用完 assignTimestampsAndWatermarks 方法后，得到的 DataStream 中的数据跟原来是一样的
        SingleOutputStreamOperator<String> dataWithWatermark = line.assignTimestampsAndWatermarks(
                new BoundedOutOfOrdernessTimestampExtractor<String>(
                        //允许乱序的延迟时间
                        Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                //提取数据中的时间
                return Long.parseLong(element.split(",")[0]);
            }
        });
        //1677168000,1 -> 1
        SingleOutputStreamOperator<Integer> nums = dataWithWatermark.map(new MapFunction<String, Integer>() {
            @Override
            public Integer map(String value) throws Exception {
                return Integer.parseInt(value.split(",")[1]);
            }
        });

        /*旧API*/
        AllWindowedStream<Integer, TimeWindow> windowed = nums.timeWindowAll(Time.seconds(5));
        windowed.sum(0).print();

        env.execute();
    }
}
